package com.navinfo.platform.basicdatastatistics.service.day;


import com.navinfo.platform.basicdatastatistics.dto.BasicDataDto;
import com.navinfo.platform.basicdatastatistics.dto.ExDataDto;
import com.navinfo.platform.basicdatastatistics.utils.day.VehicleBasicDataSchema;
import com.navinfo.platform.basicdatastatistics.utils.day.VehicleBasicDataUtil;
import com.navinfo.platform.basicdatastatistics.utils.day.VehicleExDataSchema;
import com.navinfo.platform.basicdatastatistics.utils.day.VehicleExDataUtil;
import com.navinfo.platform.common.service.ICommonService;
import com.navinfo.platform.common.service.IDataHandleService;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.StructType;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;


public enum BasicDataHandleService implements IDataHandleService {
    //
    INS;

    @Override
    public void handle(List<Object> obj, Map<String, String> configMap, SparkSession spark) {
        String day = configMap.get(ICommonService.DAY_VALUE);
        Dataset<Row> baiscDataset = castValue(obj.get(0));
        JavaRDD<Row> saveDataRDD = baiscDataset.toJavaRDD().mapPartitions(rowToBasicDataDto(day));
        StructType schema = VehicleBasicDataSchema.createSchema();
        Dataset<Row> saveDataset = spark.createDataFrame(saveDataRDD, schema);
        BasicDataService.saveData(spark, saveDataset, configMap, "hdfs.vehicle.day.save.tableName", day, schema);

        // 扩展数据
        Dataset<Row> exDataset = castValue(obj.get(1));
        StructType schemaEx = VehicleExDataSchema.createSchema();
        JavaRDD<Row> saveExDataRdd = exDataset.toJavaRDD().mapPartitions(it -> {
            List<Row> rows = new ArrayList<>();
            while (it.hasNext()) {
                Row row = it.next();
                ExDataDto dto = VehicleExDataUtil.parseRowToExDataDto(row);
                dto.setDay(Integer.parseInt(day));
                rows.add(VehicleExDataSchema.createRow(dto, schemaEx));
            }
            return rows.iterator();
        });
        Dataset<Row> saveExDataset = spark.createDataFrame(saveExDataRdd, schemaEx);
        BasicDataService.saveData(spark, saveExDataset, configMap, "hdfs.vehicle.ex.day.save.tableName", day, schemaEx);
    }

    @SuppressWarnings("unchecked")
    public static FlatMapFunction<Iterator<Row>, Row> rowToBasicDataDto(final String day) {
        return itor -> {
            List<Row> basicDataDtos = new ArrayList<>();
            StructType structType = VehicleBasicDataSchema.createSchema();
            while (itor.hasNext()) {
                Row row = itor.next();
                BasicDataDto basicDataDto = VehicleBasicDataUtil.parseRowToBasicDataDto(row);
                basicDataDto.setDay(Integer.parseInt(day));
                basicDataDtos.add(VehicleBasicDataSchema.createRow(basicDataDto, structType));
            }
            return basicDataDtos.iterator();
        };
    }
}
